// Copyright (C) 2010 and onwards Harik Shazeer Labs
// 
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// 
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// Authors: Noam Shazeer and Georges Harik

#include "objectstore.h"
#include <strstream>

ObjectStore::ObjectStore(string fn) 
  : file_(fn) {
  G_ALLOC.AddDependant(this);
  object_count_ = 0;
}

Object * ObjectStore::HashToObject(uint64 h) {
  Object ** look = hash_object_ % h;
  CHECK(look);
  return *look;
}

void ObjectStore::Read(ThreadAllocator *ta) {
  file_.Read();  
  uint64 count = 0;
  uint64 total_records = file_.RecordsAvailable();
  while (file_.RecordsAvailable()) {
    if (!(count & (count-1)))
      cout << "Reading records " << count << " / " << total_records << endl;
    count++;
    string s = file_.PopRecord();    
    const char *p = s.c_str();
    const char * endp = s.c_str() + s.size();
    MessageType mt = ReadType<MessageType>(&p);
    Object *o;
    if (mt == UNBOXED_OBJECT_MESSAGE) {
      istrstream istr(p);
      o = ReadObject(ta, istr);
    }
    else {
      CHECK(mt == ARRAY_MESSAGE);
      vector<Object *> v;
      while (p != endp) {
	uint64 h = ReadType<uint64>(&p);
	v.push_back(HashToObject(h));
      }
      o = ta->MakeArray(v);
    }
    InsertNewObject(o);
  }
}

void ObjectStore::WriteObject(Object *o) {
  CHECK(o);
  if (hash_object_ % Hash(o)) return;
  // cout << "WriteObject " << ToString(o) << " " << Hash(o) << endl;
  CHECK(!IsMutable(o));
  ObjectType t = GetType(o);
  if (IsUnboxed(o)) {
    file_.WriteRecord(RawString(UNBOXED_OBJECT_MESSAGE) 
		      + ToString(o));
  } else {
    CHECK (t == ARRAY_TYPE);
    for (uint64 i=0; i< o->ArraySize(); i++) {
      WriteObject(o->Ref(i));
    }
    string rec = RawString(ARRAY_MESSAGE);
    for (uint64 i=0; i<o->ArraySize(); i++) {
      rec += RawString(Hash(o->Ref(i)));
    }
    file_.WriteRecord(rec);
  }
  InsertNewObject(o);
}

void ObjectStore::FlagReferences(int32 shard, Frontier *frontier) {
  forall(run, hash_object_) frontier->AddObject(run->second);
}
void ObjectStore::ObjectsDeleted(int32 shard) {
  forall(run, hash_object_) run->second = NewLocation(run->second);
}

void ObjectStore::InsertNewObject(Object *o) {
  // cout << "InsertNewObject " << ToString(o) << " " << Hash(o) << endl;
  CHECK(!(hash_object_ % Hash(o)));
  hash_object_[Hash(o)] = o;
  object_count_++;
}
